"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : python驱动psycopg2
Case Name   : 配置白名单，md5方式连接数据库
Create At   : 2023/9/21
Owner       : opentestcase022
Description :
    1.解压openGauss-*-CentOS-x86_64-Python.tar.gz得到lib,替换psycopg2
    2.修改加密方式为0
    3.新建用户并赋权
    4.配置白名单
    5.配置lib
    6.连接数据库
Expect      :
    1.拷贝成功
    2.执行成功
    3.执行成功
    4.配置成功
    5.配置成功
    6.连接数据库成功
History     :
    2023.9.22 取消连接信息中对密码的校验适配不同操作系统
"""
import os
import unittest

from yat.test import macro

from testcase.CONNECTORS.PDBC.PSYCOPG2. \
    Opengauss_Function_Connect_Psycopg_Common import Psycopg2Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class Psycopg5(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        text = f'-----{os.path.basename(__file__)} start-----'
        self.log.info(text)
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.Constant = Constant()
        self.user = 'u_psycopg5'
        self.u_pwd = macro.COMMON_PASSWD
        self.conn_info = {'dbname': self.pri_sh.node.db_name,
                          'user': self.user,
                          'password': self.u_pwd,
                          'host': self.pri_sh.node.ssh_host,
                          'port': self.pri_sh.node.db_port}
        self.psycopg_com = Psycopg2Common(**self.conn_info)

        text = '----前置操作: 备份白名单 expect: 备份成功----'
        self.log.info(text)
        res = self.psycopg_com.bak_conf(bak_flag=True)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

    def test_1(self):
        text = '----step1: 解压PDBC得到lib,替换psycopg2 expect: 执行成功----'
        res = self.psycopg_com.decompress_pkg()
        self.assertTrue(res, f'执行失败: {text}')
        res = self.psycopg_com.set_new_psycopg2()
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step2: 修改加密方式为0 expect: 配置成功----'
        self.log.info(text)
        res = self.pri_sh.execute_gsguc('reload',
                                        self.Constant.GSGUC_SUCCESS_MSG,
                                        'password_encryption_type=0')
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step3: 新建用户并赋权 expect: 配置成功----'
        self.log.info(text)
        sql = f'create user {self.user} identified by \\\"{self.u_pwd}\\\";' \
            f'grant all privileges on database ' \
            f'{self.conn_info["dbname"]} to {self.user}'
        res = self.pri_sh.execut_db_sql(sql)
        self.assertIn(self.Constant.CREATE_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.Constant.GRANT_SUCCESS_MSG, res, f'执行失败: {text}')

        text = '----step4: 配置白名单 expect: 配置成功----'
        self.log.info(text)
        res = self.psycopg_com.set_pghba(open_flag=True,
                                         db=self.conn_info["dbname"],
                                         user=self.user,
                                         mode='md5')
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step5: 配置lib expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.set_lib_path()
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step6: 连接数据库 expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.conn_db()
        expect = f"user={self.user} " \
            f"dbname={self.pri_sh.node.db_name} " \
            f"host={self.pri_sh.node.ssh_host} " \
            f"port={self.pri_sh.node.db_port}"
        res = [self.assertIn(i, res) for i in expect.split()]
        self.assertEqual(len(res), 4, f'执行失败: {text}')

    def tearDown(self):
        text = '----run teardown----'
        self.log.info(text)

        text1 = f'----后置操作：删除解压文件目录----'
        self.log.info(text1)
        res1 = self.psycopg_com.clean_decompress_file()

        text2 = f'----后置操作：还原白名单----'
        self.log.info(text2)
        res2 = self.psycopg_com.bak_conf(bak_flag=False)

        text3 = f'----后置操作：删除用户----'
        self.log.info(text3)
        sql = f"drop user {self.user} cascade;"
        res3 = self.pri_sh.execut_db_sql(sql)

        text4 = f'----后置操作：还原数据库加密方式----'
        self.log.info(text4)
        res4 = self.pri_sh.execute_gsguc('reload',
                                         self.Constant.GSGUC_SUCCESS_MSG,
                                         'password_encryption_type=2')

        self.assertTrue(res1, f'执行失败: {text1}')
        self.assertTrue(res2, f'执行失败: {text2}')
        self.assertIn(self.Constant.DROP_ROLE_SUCCESS_MSG, res3,
                      f'执行失败: {text3}')
        self.assertTrue(res4, f'执行失败: {text4}')

        text = f'-----{os.path.basename(__file__)} end-----'
        self.log.info(text)
